package wsf.akkascala

import akka.actor.{Actor, ActorRef, Props, UntypedActor}
import akkascala.{MapData, ReduceData}

/**
 * Created by sjzx on 2016/5/13.
 */
class MasterActor extends Actor {
  var mainSender:ActorRef = null;
  val aggregateActor = context.actorOf(Props(classOf[AggreateWorker]), "aggregateworker")// create aggregate actor,this will be the arg of reduce actor
  val reduceActor = context.actorOf(Props(classOf[ReduceWorker], aggregateActor), "reduceworker")//create reduce actor,  Props(classOf[ReduceWorker], aggregateActor)
  val mapActor: ActorRef = context.actorOf(Props(classOf[MapWorker], reduceActor), "mapworker")
  println(aggregateActor.path)  //akka://akkatest/user/akkaservice/aggregateworker  the path at the end of akkaservice which create by system
  override def receive={
    case "res" => aggregateActor ! "res"; mainSender = sender
    case "test" => println("test")
    case "become" =>
      println("become")
      self ! "test"
      context.become{//会将当前receive里的内容替换成become快里面的内容
        case "unbecome" => println("unbecome")
        case "test" => println("test")
          context.unbecome()
      }
    case s : String => mapActor ! s
    case message : MapData => reduceActor ! message
    case message : ReduceData => mainSender ! message
    case _ => unhandled("Error")
  }

//  context.stop(self) //use this method to stop actor self!
  /**
    * when the actor to be stopped, this method will be doing
    */
  override def postStop={
    println("stop!")
  }
}
